package com.mxl.rabbitmq.consumer;

import com.alibaba.fastjson.JSON;
import com.mxl.rabbitmq.config.JavaBoyConfig;
import com.mxl.rabbitmq.model.MessageBody;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * 消费者 从消息队列获取消息
 * @author MXL
 * @date 2021/9/14 0014
 **/
@Component
@RabbitListener(queues = JavaBoyConfig.JAVABOY_QUEUE_NAME/*, queuesToDeclare = @Queue(JavaBoyConfig.JAVABOY_QUEUE_NAME)*/)
@Slf4j
public class JavaBoyReceiver {

    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * 默认情况下，消息消费方法自带事务，即如果该方法在执行过程中抛出异常，
     * 那么被消费的消息会重新回到队列中等待下一次被消费，如果该方法正常执行完没有抛出异常，则这条消息就算是被消费了。
     *
     * @param message
     */
//    @RabbitHandler
    public void process(Message message) {
        //handleByAuto(message, channel);
        log.info("JavaBoy消息消费者接收到消息: {}", JSON.toJSONString(message));
    }

//    @RabbitHandler
    public void process2(String msg) {
        //handleByAuto(message, channel);
        log.info("JavaBoy消息消费者接收到消息: {}", msg);
    }

    @RabbitHandler
    public void process3(Message message, MessageBody messageBody, Channel channel) {
        //自动确认
//        handleByAuto(messageBody, channel);
        //手动确认
        handleByHand(messageBody, message, channel);
    }

    /**
     * 自动确认
     *
     * @param message
     * @param channel
     */
    private void handleByAuto(MessageBody message, Channel channel) {
        //获取消息编号
//        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        long deliveryTag = 0;
        try {
            //拒绝消息
            //第二个参数是 requeue，即是否重新入队。
            //如果第二个参数为 true，则这条被拒绝的消息会重新进入到消息队列中，等待下一次被消费；如果第二个参数为 false，则这条被拒绝的消息就会被丢掉，不会有新的消费者去消费它了。
            //需要注意的是，basicReject 方法一次只能拒绝一条消息
//            channel.basicReject(deliveryTag, true);
//            log.error("拒绝消息编号: ", deliveryTag);

            //消息消费代码写在这里

            //消费完成后, 手动 ack
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
        }
        //默认情况下，消息消费方法自带事务，即如果该方法在执行过程中抛出异常，那么被消费的消息会重新回到队列中等待下一次被消费，
        //如果该方法正常执行完没有抛出异常，则这条消息就算是被消费了
        //int i = 1 / 0;
    }

    /**
     * 推模式手动确认
     *
     * @param message
     * @param channel
     */
    private void handleByHand(MessageBody messageBody, Message message, Channel channel) {
//        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        MessageHeaders headers = message.getHeaders();
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            //消息消费的代码写到这里
            log.info("推模式(默认模式)-手动确认-messageBody:{} ", messageBody);
            //消费完成后，手动 ack
            /**
             * basicAck：这个是手动确认消息已经成功消费，该方法有两个参数：
             *  第一个参数表示消息的 id；
             *
             *  第二个参数 multiple:
             *      如果为 false，表示仅确认当前消息消费成功，如果为 true，则表示当前消息之前所有未被当前消费者确认的消息都消费成功。
             */
            //只手动确认后10个
            if (messageBody.getIndex() >= 0 ){
                channel.basicAck(deliveryTag, false);
            }
        } catch (Exception e) {
            //手动 nack
            try {
                /**
                 * basicNack：这个是告诉 RabbitMQ 当前消息未被成功消费，
                 * 该方法有三个参数：
                 * 第一个参数表示消息的 id；
                 * 第二个参数 multiple:
                 *      如果为 false，表示仅拒绝当前消息的消费，如果为 true，则表示拒绝当前消息之前所有未被当前消费者确认的消息；
                 * 第三个参数 requeue:
                 *      含义和前面所说的一样，被拒绝的消息是否重新入队
                 */
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }

    /**
     * 拉模式手动确认
     */
    private void handleByHandAndPull() {
        Channel channel = rabbitTemplate.getConnectionFactory().createConnection().createChannel(true);
        long deliveryTag = 0L;
        try {
            GetResponse getResponse = channel.basicGet(JavaBoyConfig.JAVABOY_QUEUE_NAME, false);
            deliveryTag = getResponse.getEnvelope().getDeliveryTag();
            System.out.println("o = " + new String((getResponse.getBody()), "UTF-8"));
            channel.basicAck(deliveryTag, false);
        } catch (IOException e) {
            try {
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
    }
}
